package org.example.lock;

import com.alibaba.ttl.TransmittableThreadLocal;
import org.apache.zookeeper.*;
import org.apache.zookeeper.data.Stat;
import org.slf4j.Logger;
import org.slf4j.LoggerFactory;
import org.springframework.stereotype.Component;

import javax.annotation.PostConstruct;
import javax.annotation.Resource;
import java.nio.charset.StandardCharsets;
import java.util.Collections;
import java.util.List;
import java.util.concurrent.CountDownLatch;

/**
 * @author shenguangyang
 */
@Component
public class ZkLock extends LockTemplate {
    private static final Logger log = LoggerFactory.getLogger(ZkLock.class);
    protected static final String DEFAULT_LOCKS_PATH = "/locks";
    protected String lockPath;

    private static final TransmittableThreadLocal<String> currentNodePathTtl = new TransmittableThreadLocal<>();

    @Resource
    protected ZooKeeper zkClient;

    @PostConstruct
    public void init() {
        this.lockPath = DEFAULT_LOCKS_PATH;
        createNodeLocks();
    }

    private void createNodeLocks() {
        try {
            Stat exists = zkClient.exists(lockPath, false);
            if (exists == null) {
                zkClient.create(lockPath, "locks".getBytes(StandardCharsets.UTF_8), ZooDefs.Ids.OPEN_ACL_UNSAFE, CreateMode.PERSISTENT);
            }
        } catch (KeeperException | InterruptedException e) {
            log.error("create locks node fail, error: {}", e.getMessage());
            throw new RuntimeException(e);
        }
    }

    @Override
    public void unlock() {
        try {
            String currentNodePath = currentNodePathTtl.get();
            if (currentNodePath == null || "".equals(currentNodePath)) {
                return;
            }
            Stat exists = zkClient.exists(currentNodePath, false);
            if (exists != null) {
                zkClient.delete(currentNodePath, -1);
            }
            log.debug("unlock --- delete node path: {}", currentNodePath);
        } catch (Exception e) {
            log.error("unlock fail, error: ", e);
        }
    }

    @Override
    protected boolean tryLock() {
        try {
            // 创建对应的临时带序号节点
            String currentNode = zkClient.create(this.lockPath + "/" + "seq-", null, ZooDefs.Ids.OPEN_ACL_UNSAFE, CreateMode.EPHEMERAL_SEQUENTIAL);
            // 如果children只有一个值, 那就直接获取锁, 如果有多个节点, 需要判断谁最小
            List<String> children = zkClient.getChildren(this.lockPath, false);

            // 判断创建的节点是否是最小的序号节点, 如果 是获取到锁; 如果不是, 监听他序号的前一个节点
            if (children.size() == 1) {
                currentNodePathTtl.set(currentNode);
                return true;
            } else {
                Collections.sort(children);

                // 获取节点名称 seq-00000000
                String thisNode = currentNode.substring((this.lockPath + "/").length());
                int index = children.indexOf(thisNode);
                String waitNodePath;
                if (index == -1) {
                    throw new RuntimeException("数据异常");
                } else if (index == 0) {
                    // 是第一个节点直接获取锁
                    currentNodePathTtl.set(currentNode);
                    return true;
                } else {
                    // 监听前一个节点
                    CountDownLatch waitLath = new CountDownLatch(1);
                    waitNodePath = this.lockPath + "/" + children.get(index - 1);
                    zkClient.getData(waitNodePath, watchedEvent -> {
                        if (watchedEvent.getType() == Watcher.Event.EventType.NodeDeleted && watchedEvent.getPath().equals(waitNodePath)) {
                            waitLath.countDown();
                        }
                    }, null);
                    waitLath.await();
                    currentNodePathTtl.set(currentNode);
                }
            }
            return true;
        } catch (KeeperException | InterruptedException e) {
            // 说明前一个节点已经释放了锁
            if (e instanceof KeeperException.NoNodeException) {
                return true;
            }
            log.error("try lock fail, error: ", e);
        }
        return false;
    }

    @Override
    protected void waitLock() {

    }
}
